package com.surfilter.datax.plugin.reader.clickhousereader;

import ch.qos.logback.core.joran.spi.ElementSelector;
import com.alibaba.datax.common.element.*;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.reader.Key;
import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.ClickHouseStatement;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;

public class ClickhouseReader extends Reader {

    private static final Logger LOG = LoggerFactory.getLogger(ClickhouseReader.class);


    public static class Job extends Reader.Job {

        private Configuration originalConfig = null;

        /**
         * 执行第1步
         */
        @Override
        public void preCheck() {
            super.preCheck();
        }

        /**
         * 执行第2步
         *
         * @param jobConfiguration
         */
        @Override
        public void preHandler(Configuration jobConfiguration) {
            super.preHandler(jobConfiguration);
        }

        /**
         * 执行第3步
         * 根据configuration来初始化reader和writer插件，同时设置reader和writer的configuration信息
         */
        @Override
        public void init() {
            //获取配置文件
            this.originalConfig = super.getPluginJobConf();

        }

        /**
         * 执行第4步
         * 初始化reader和writer插件的初始化，通过调用插件的prepare()方法实现，每个插件都有自己的jarLoader，通过集成URLClassloader实现而来
         */
        @Override
        public void prepare() {
            super.prepare();
        }

        /**
         * 执行第5步
         *
         * @param adviceNumber 着重说明下，adviceNumber是框架建议插件切分的任务数，插件开发人员最好切分出来的任务数>=
         *                     adviceNumber。<br>
         *                     <br>
         *                     之所以采取这个建议是为了给用户最好的实现，例如框架根据计算认为用户数据存储可以支持100个并发连接，
         *                     并且用户认为需要100个并发。 此时，插件开发人员如果能够根据上述切分规则进行切分并做到>=100连接信息，
         *                     DataX就可以同时启动100个Channel，这样给用户最好的吞吐量 <br>
         *                     例如用户同步一张Mysql单表，但是认为可以到10并发吞吐量，插件开发人员最好对该表进行切分，比如使用主键范围切分，
         *                     并且如果最终切分任务数到>=10，我们就可以提供给用户最大的吞吐量。 <br>
         *                     <br>
         *                     当然，我们这里只是提供一个建议值，Reader插件可以按照自己规则切分。但是我们更建议按照框架提供的建议值来切分。 <br>
         *                     <br>
         *                     对于ODPS写入OTS而言，如果存在预排序预切分问题，这样就可能只能按照分区信息切分，无法更细粒度切分，
         *                     这类情况只能按照源头物理信息切分规则切分。 <br>
         *                     <br>
         * @return
         */
        @Override
        public List<Configuration> split(int adviceNumber) {
            List<Configuration> readerSplitConfigurations = new ArrayList<Configuration>();
            for (int i = 0; i < adviceNumber; i++) {
                Configuration readerSplitConfiguration = this.originalConfig.clone();
                readerSplitConfigurations.add(readerSplitConfiguration);
            }
            return readerSplitConfigurations;
        }

        /**
         * 执行第6步
         */
        @Override
        public void post() {
            super.post();
        }

        /**
         * 执行第7步
         *
         * @param jobConfiguration
         */
        @Override
        public void postHandler(Configuration jobConfiguration) {
            super.postHandler(jobConfiguration);
        }


        /**
         * 执行第8步
         */
        @Override
        public void destroy() {
        }


    }

    public static class Task extends Reader.Task {

        String username;
        String password;
        String url;
        String table;
        ClickHouseDataSource dataSource;
        ClickHouseConnection connection;
        ClickHouseStatement statement;


        @Override
        public void startRead(RecordSender recordSender) {
            ResultSet resultSet = null;
            int  columnNumber ;
            try {
                resultSet = statement.executeQuery("select * from s_jt_hcprghyxx_test");
                ResultSetMetaData rsmd = resultSet.getMetaData();
                columnNumber = rsmd.getColumnCount();
                List<Record> list = new ArrayList();
                while (resultSet.next()) {
                    Record record;
                    for (int i = 1; i <= rsmd.getColumnCount(); i++) {
                        record = buildRecord(recordSender,resultSet,rsmd,columnNumber,super.getTaskPluginCollector());
                        //进行验证，如果验证成功，发送record，否则发送至脏数据
                        //TODO 这里识别是否为验证未通过数据
                        boolean validate=true;
                        if(validate){
                            recordSender.sendToWriter(record);
                        }else{
                            //TODO:这里要考虑如何写库
                            super.getTaskPluginCollector().collectDirtyRecord(record, new Exception("验证失败"));
                        }

                    }
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void init() {
            //获取用户名密码以及数据库连接url信息
            this.username = this.getPluginJobConf().getNecessaryValue(Key.USERNAME, DBUtilErrorCode.REQUIRED_VALUE);
            this.password = this.getPluginJobConf().getNecessaryValue(Key.PASSWORD, DBUtilErrorCode.REQUIRED_VALUE);
            this.url = this.getPluginJobConf().getNecessaryValue(Key.JDBC_URL, DBUtilErrorCode.REQUIRED_VALUE);
            this.table = this.getPluginJobConf().getNecessaryValue(Key.TABLE, DBUtilErrorCode.REQUIRED_VALUE);
            Properties props = new Properties();
            dataSource = new ClickHouseDataSource(url, props);
            try {
                connection = dataSource.getConnection();
                statement = connection.createStatement();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void destroy() {
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    protected static Record buildRecord(RecordSender recordSender, ResultSet rs, ResultSetMetaData metaData, int columnNumber, TaskPluginCollector taskPluginCollector) {
        Record record = recordSender.createRecord();
        try {
            //根据数据类型进行数据封装
            for (int i = 1; i <= columnNumber; i++) {
                switch (metaData.getColumnType(i)) {
                    case Types.CHAR:
                    case Types.NCHAR:
                    case Types.VARCHAR:
                    case Types.LONGVARCHAR:
                    case Types.NVARCHAR:
                    case Types.CLOB:
                    case Types.NCLOB:
                        record.addColumn(new StringColumn(rs.getString(i)));
                        break;

                    case Types.SMALLINT:
                    case Types.TINYINT:
                    case Types.INTEGER:
                    case Types.BIGINT:
                        record.addColumn(new LongColumn(rs.getString(i)));
                        break;

                    case Types.NUMERIC:
                    case Types.DECIMAL:
                        record.addColumn(new DoubleColumn(rs.getString(i)));
                        break;

                    case Types.FLOAT:
                    case Types.REAL:
                    case Types.DOUBLE:
                        record.addColumn(new DoubleColumn(rs.getString(i)));
                        break;

                    case Types.TIME:
                        record.addColumn(new DateColumn(rs.getTime(i)));
                        break;

                    // for mysql bug, see http://bugs.mysql.com/bug.php?id=35115
                    case Types.DATE:
                        if (metaData.getColumnTypeName(i).equalsIgnoreCase("year")) {
                            record.addColumn(new LongColumn(rs.getInt(i)));
                        } else {
                            record.addColumn(new DateColumn(rs.getDate(i)));
                        }
                        break;

                    case Types.TIMESTAMP:
                        record.addColumn(new DateColumn(rs.getTimestamp(i)));
                        break;

                    case Types.BINARY:
                    case Types.VARBINARY:
                    case Types.BLOB:
                    case Types.LONGVARBINARY:
                        record.addColumn(new BytesColumn(rs.getBytes(i)));
                        break;

                    // warn: bit(1) -> Types.BIT 可使用BoolColumn
                    // warn: bit(>1) -> Types.VARBINARY 可使用BytesColumn
                    case Types.BOOLEAN:
                    case Types.BIT:
                        record.addColumn(new BoolColumn(rs.getBoolean(i)));
                        break;

                    case Types.NULL:
                        String stringData = null;
                        if (rs.getObject(i) != null) {
                            stringData = rs.getObject(i).toString();
                        }
                        record.addColumn(new StringColumn(stringData));
                        break;

                    default:
                        throw DataXException
                                .asDataXException(
                                        DBUtilErrorCode.UNSUPPORTED_TYPE,
                                        String.format(
                                                "您的配置文件中的列配置信息有误. 因为DataX 不支持数据库读取这种字段类型. 字段名:[%s], 字段名称:[%s], 字段Java类型:[%s]. 请尝试使用数据库函数将其转换datax支持的类型 或者不同步该字段 .",
                                                metaData.getColumnName(i),
                                                metaData.getColumnType(i),
                                                metaData.getColumnClassName(i)));
                }

            }

            //收集验证错误的数据，集中进行处理

        } catch (Exception e) {
            LOG.debug("read data " + record.toString()
                    + " occur exception:", e);
            //TODO 这里识别为脏数据靠谱吗？不靠谱
            taskPluginCollector.collectDirtyRecord(record, e);
            if (e instanceof DataXException) {
                throw (DataXException) e;
            }
        }
        return record;
    }
}